package com.yanqu.data.analytics.impl;

import cn.thinkingdata.tga.javasdk.BatchConsumer;
import cn.thinkingdata.tga.javasdk.DebugConsumer;
import cn.thinkingdata.tga.javasdk.LoggerConsumer;
import cn.thinkingdata.tga.javasdk.ThinkingDataAnalytics;
import cn.thinkingdata.tga.javasdk.exception.InvalidArgumentException;
import cn.thinkingdata.tga.javasdk.inter.Consumer;
import com.yanqu.data.analytics.Provider;
import com.yanqu.data.analytics.ThinkingInnerConsumer;
import com.yanqu.data.analytics.consumer.CollectConsumer;
import com.yanqu.data.analytics.consumer.ConsumerQueueOperate;
import com.yanqu.data.analytics.consumer.KafKaProductConsumer;
import com.yanqu.road.utils.ConfigHelper;
import com.yanqu.road.utils.MyProperties;
import com.yanqu.road.utils.executor.GeneralThreadFactory;
import com.yanqu.road.utils.string.StringUtils;
import org.apache.logging.log4j.Logger;

import java.net.URISyntaxException;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThinkingProvider implements Provider {

    Logger logger;

    MyProperties properties;

    ThinkingDataAnalytics tda;

    String serverUrl;

    //独立线程
    ExecutorService executorService;

    //访问队列的入口
    ConsumerQueueOperate queueOperate;

    public ThinkingProvider(Logger logger, String serverUrl) {
        this.logger = logger;
        this.serverUrl = serverUrl;
        executorService = Executors.newFixedThreadPool(2, new GeneralThreadFactory("tda-pool"));
    }

    public void init() throws Exception{
        String path = ConfigHelper.getValue("tda.path");

        //沒有tda的相关配置
        if(StringUtils.isNullOrEmpty(path)){
            initCollectConsumerTda();
            return;
        }

        try{
            this.properties = new MyProperties(path);

            String consumer = properties.getProperty("consumer");
            if("log".equals(consumer)){
                initLogConsumerTda();
            }else if("batch".equals(consumer)){
                initBatchConsumerTda();
            }else if("debug".equals(consumer)){
                initDebugConsumerTda();
            }else if("inner".equals(consumer)){
                ThinkingInnerConsumer innerConsumer = new ThinkingInnerConsumer(logger);
                tda = new ThinkingDataAnalytics(innerConsumer, true);
            }else if("collect".equals(consumer)){
                initCollectConsumerTda();
            }else if("kafka".equals(consumer)){
                initKafkaProductConsumer();
            }

        }catch (Exception e){
            //tda配置读取不到，直接采用直连方式
            initCollectConsumerTda();
        }

        logger.debug(" init ThinkingProvider ");
    }

    /**
     * 集中收集
     */
    private void initCollectConsumerTda() {
        CollectConsumer collectConsumer = new CollectConsumer(logger);
        //默认加serverId
        String serverId = ConfigHelper.getPath("serverId");
        if(serverUrl.contains("?")){
            serverUrl += "&serverId=" + serverId;
        }else {
            serverUrl += "?serverId=" + serverId;
        }
        collectConsumer.setServerUrl(serverUrl);
        queueOperate = collectConsumer;
        tda = new ThinkingDataAnalytics(collectConsumer, true);
        logger.info("initCollectConsumerTda serverUrl:{}", serverUrl);
    }

    protected void initKafkaProductConsumer(){
        long serverId = ConfigHelper.getLong("serverId");
        KafKaProductConsumer consumer = new KafKaProductConsumer(logger, serverUrl, serverId);
        queueOperate = consumer;
        tda = new ThinkingDataAnalytics(consumer, true);
        logger.info("initKafkaProductConsumer serverUrl:{}", serverUrl);
    }

    /**
     * 直连-单条-测试模式，数据不进入项目，可在平台查看debug校验
     */
    private void initDebugConsumerTda() throws URISyntaxException {
        String appId = properties.getProperty("appId");
        String serverUrl = properties.getProperty("serverUrl");
        Consumer consumerContext = new DebugConsumer(serverUrl,appId);
        tda = new ThinkingDataAnalytics(consumerContext);
        logger.info("initDebugConsumerTda serverUrl:{}", serverUrl);
    }

    /**
     * 直连-批量模式
     */
    public void initBatchConsumerTda() throws URISyntaxException {
        String appId = properties.getProperty("appId");
        String serverUrl = properties.getProperty("serverUrl");
        BatchConsumer.Config config = new BatchConsumer.Config();
        config.setAutoFlush(properties.getBoolean("autoFlush"));
        config.setInterval(properties.getInt("interval"));
        config.setThrowException(properties.getBoolean("isThrowException"));
        config.setCompress(properties.getProperty("compress"));
        config.setTimeout(properties.getInt("timeout"));
        config.setBatchSize(properties.getInt("batchSize", 1000));
        config.setMaxCacheSize(properties.getInt("maxCacheSize"));
        Consumer consumerContext = new BatchConsumer(serverUrl, appId, config);
        tda = new ThinkingDataAnalytics(consumerContext);
        logger.info("initBatchConsumerTda serverUrl:{}", serverUrl);
    }

    /**
     * 直接写入本地文件（按照文日期周期、文件大小滚动存储）
     */
    public void initLogConsumerTda() {
        //默认加serverId
        String serverId = ConfigHelper.getPath("serverId");
        //根目录不加serverId
        String logDirectory = properties.getProperty("logDirectory");
        //写文件锁
        String lockFileName = properties.getProperty("lockFileName") + "_" + serverId;
        //及文件需加serverId
        String fileNamePrefix = properties.getProperty("fileNamePrefix") + "." + serverId;

        int fileSize = properties.getInt("fileSize");

        LoggerConsumer.Config config = new LoggerConsumer.Config(logDirectory, fileSize);
        String rotateMode = properties.getProperty("rotateMode", "DAILY");
        config.setRotateMode("DAILY".equals(rotateMode)? LoggerConsumer.RotateMode.DAILY : LoggerConsumer.RotateMode.HOURLY);
        config.setAutoFlush(properties.getBoolean("autoFlush"));
        config.setInterval(properties.getInt("interval"));
        config.setFilenamePrefix(fileNamePrefix);
        config.setFileSize(properties.getInt("fileSize", 1024));
        config.setBufferSize(properties.getInt("bufferSize", 8192));
        config.setLockFile(lockFileName);
        Consumer consumerContext = new LoggerConsumer(config);
        tda = new ThinkingDataAnalytics(consumerContext);
        logger.info("initBatchConsumerTda logDirectory:{}", logDirectory);
    }

    @Override
    public void user_del(String accountId, String distinctId) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.user_del(accountId, distinctId);
                } catch (InvalidArgumentException e) {
                    logger.error("accountId {} distinctId{}", accountId, distinctId, e);
                }
            }
        });
    }

    @Override
    public void user_add(String accountId, String distinctId, Map<String, Object> properties) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.user_add(accountId, distinctId, properties);
                } catch (InvalidArgumentException e) {
                    logger.error("accountId {} distinctId{}", accountId, distinctId, e);
                }
            }
        });
    }

    @Override
    public void user_setOnce(String accountId, String distinctId, Map<String, Object> properties) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.user_setOnce(accountId, distinctId, properties);
                } catch (InvalidArgumentException e) {
                    logger.error("accountId {} distinctId{}", accountId, distinctId, e);
                    for (Map.Entry<String, Object> entry : properties.entrySet()) {
                        logger.error("key:{}, value:{}",entry.getKey(), entry.getValue());
                    }
                }
            }
        });
    }

    @Override
    public void user_set(String accountId, String distinctId, Map<String, Object> properties) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.user_set(accountId, distinctId, properties);
                } catch (InvalidArgumentException e) {
                    logger.error("accountId {} distinctId{}", accountId, distinctId, e);
                }
            }
        });
    }

    @Override
    public void user_unset(String accountId, String distinctId, String... properties) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.user_unset(accountId, distinctId, properties);
                } catch (InvalidArgumentException e) {
                    logger.error("accountId {} distinctId{}", accountId, distinctId, e);
                }
            }
        });
    }

    @Override
    public void user_append(String accountId, String distinctId, Map<String, Object> properties) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.user_append(accountId, distinctId, properties);
                } catch (InvalidArgumentException e) {
                    logger.error("accountId {} distinctId{}", accountId, distinctId, e);
                }
            }
        });
    }

    @Override
    public void track(String accountId, String distinctId, String eventName, Map<String, Object> properties) {
        if(properties == null){
            logger.info("track accountId {} distinctId{} eventName {} properties null", accountId, distinctId, eventName);
            return;
        }
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.track(accountId, distinctId, eventName, properties);
                } catch (Exception e) {
                    logger.error("track accountId {} distinctId{} eventName {}", accountId, distinctId, eventName, e);
                    for (Map.Entry<String, Object> entry : properties.entrySet()) {
                        logger.error("key:{}, value:{}",entry.getKey(), entry.getValue());
                    }
                }
            }
        });
    }

    @Override
    public void track_update(String accountId, String distinctId, String eventName, String eventId, Map<String, Object> properties) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.track_update(accountId, distinctId, eventName, eventId, properties);
                } catch (InvalidArgumentException e) {
                    logger.error("track_update accountId {} distinctId{} eventName {} eventId {}", accountId, distinctId, eventName, eventId, e);
                }
            }
        });
    }

    @Override
    public void track_overwrite(String accountId, String distinctId, String eventName, String eventId, Map<String, Object> properties) {
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    tda.track_overwrite(accountId, distinctId, eventName, eventId, properties);
                } catch (InvalidArgumentException e) {
                    logger.error("track_overwrite accountId {} distinctId{} eventName {} eventId {}", accountId, distinctId, eventName, eventId, e);
                }
            }
        });
    }

    @Override
    public void flush() {
        tda.flush();
    }

    @Override
    public void close() {
        tda.close();
    }

    @Override
    public ConsumerQueueOperate getQueueOperate() {
        return queueOperate;
    }

}
